Enter 2006 September/Enter 09 2006.iso/Internet/SpamExperts Home 1.1/SpamExperts Home.exe/lib/spamexperts.modules/dns/rdata.pyc

Python Compiled Bytecode | 2006-07-14 | 14.1 KB | 451 lines

# Source Generated with Decompyle++
# File: in.pyc (Python 2.4)

"""DNS rdata.

@var _rdata_modules: A dictionary mapping a (rdclass, rdtype) tuple to
the module which implements that type.
@type _rdata_modules: dict
@var _module_prefix: The prefix to use when forming modules names.  The
default is 'dns.rdtypes'.  Changing this value will break the library.
@type _module_prefix: string
@var _hex_chunk: At most this many octets that will be represented in each
chunk of hexstring that _hexify() produces before whitespace occurs.
@type _hex_chunk: int"""
import dns.exception as dns
import dns.rdataclass as dns
import dns.rdatatype as dns
import dns.tokenizer as dns
_hex_chunksize = 32

def _hexify(data, chunksize = None):
    '''Convert a binary string into its hex encoding, broken up into chunks
    of I{chunksize} characters separated by a space.

    @param data: the binary string
    @type data: string
    @param chunksize: the chunk size.  Default is L{dns.rdata._hex_chunksize}
    @rtype: string
    '''
    if chunksize is None:
        chunksize = _hex_chunksize

    hex = data.encode('hex_codec')
    l = len(hex)
    if l > chunksize:
        chunks = []
        i = 0
        while i < l:
            chunks.append(hex[i:i + chunksize])
            i += chunksize
        hex = ' '.join(chunks)

    return hex

_base64_chunksize = 32

def _base64ify(data, chunksize = None):
    '''Convert a binary string into its base64 encoding, broken up into chunks
    of I{chunksize} characters separated by a space.

    @param data: the binary string
    @type data: string
    @param chunksize: the chunk size.  Default is
    L{dns.rdata._base64_chunksize}
    @rtype: string
    '''
    if chunksize is None:
        chunksize = _base64_chunksize

    b64 = data.encode('base64_codec')
    b64 = b64.replace('\n', '')
    l = len(b64)
    if l > chunksize:
        chunks = []
        i = 0
        while i < l:
            chunks.append(b64[i:i + chunksize])
            i += chunksize
        b64 = ' '.join(chunks)

    return b64

__escaped = {
    '"': True,
    '\\': True }

def _escapify(qstring):
    '''Escape the characters in a quoted string which need it.

    @param qstring: the string
    @type qstring: string
    @returns: the escaped string
    @rtype: string
    '''
    text = ''
    for c in qstring:
        if c in __escaped:
            text += '\\' + c
            continue
        if ord(c) >= 32 and ord(c) < 127:
            text += c
            continue
        text += '\\%03d' % ord(c)

    return text


def _truncate_bitmap(what):
    """Determine the index of greatest byte that isn't all zeros, and
    return the bitmap that contains all the bytes less than that index.

    @param what: a string of octets representing a bitmap.
    @type what: string
    @rtype: string
    """
    for i in xrange(len(what) - 1, -1, -1):
        if what[i] != '\x00':
            break
            continue

    return ''.join(what[0:i + 1])


class Rdata(object):
    '''Base class for all DNS rdata types.
    '''
    __slots__ = [
        'rdclass',
        'rdtype']

    def __init__(self, rdclass, rdtype):
        '''Initialize an rdata.
        @param rdclass: The rdata class
        @type rdclass: int
        @param rdtype: The rdata type
        @type rdtype: int
        '''
        self.rdclass = rdclass
        self.rdtype = rdtype


    def covers(self):
        '''DNS SIG/RRSIG rdatas apply to a specific type; this type is
        returned by the covers() function.  If the rdata type is not
        SIG or RRSIG, dns.rdatatype.NONE is returned.  This is useful when
        creating rdatasets, allowing the rdataset to contain only RRSIGs
        of a particular type, e.g. RRSIG(NS).
        @rtype: int
        '''
        return dns.rdatatype.NONE


    def extended_rdatatype(self):
        '''Return a 32-bit type value, the least significant 16 bits of
        which are the ordinary DNS type, and the upper 16 bits of which are
        the "covered" type, if any.
        @rtype: int
        '''
        return self.covers() << 16 | self.rdtype


    def to_text(self, origin = None, relativize = True, **kw):
        '''Convert an rdata to text format.
        @rtype: string
        '''
        raise NotImplementedError


    def to_wire(self, file, compress = None, origin = None):
        '''Convert an rdata to wire format.
        @rtype: string
        '''
        raise NotImplementedError


    def __repr__(self):
        covers = self.covers()
        if covers == dns.rdatatype.NONE:
            ctext = ''
        else:
            ctext = '(' + dns.rdatatype.to_text(covers) + ')'
        return '<DNS ' + dns.rdataclass.to_text(self.rdclass) + ' ' + dns.rdatatype.to_text(self.rdtype) + ctext + ' rdata: ' + str(self) + '>'


    def __str__(self):
        return self.to_text()


    def _cmp(self, other):
        '''Compare an rdata with another rdata of the same rdtype and
        rdclass.  Return < 0 if self < other in the DNSSEC ordering,
        0 if self == other, and > 0 if self > other.
        '''
        raise NotImplementedError


    def __eq__(self, other):
        if not isinstance(other, Rdata):
            return False

        if self.rdclass != other.rdclass or self.rdtype != other.rdtype:
            return False

        return self._cmp(other) == 0


    def __ne__(self, other):
        if not isinstance(other, Rdata):
            return True

        if self.rdclass != other.rdclass or self.rdtype != other.rdtype:
            return True

        return self._cmp(other) != 0


    def __lt__(self, other):
        if not isinstance(other, Rdata) and self.rdclass != other.rdclass or self.rdtype != other.rdtype:
            return NotImplemented

        return self._cmp(other) < 0


    def __le__(self, other):
        if not isinstance(other, Rdata) and self.rdclass != other.rdclass or self.rdtype != other.rdtype:
            return NotImplemented

        return self._cmp(other) <= 0


    def __ge__(self, other):
        if not isinstance(other, Rdata) and self.rdclass != other.rdclass or self.rdtype != other.rdtype:
            return NotImplemented

        return self._cmp(other) >= 0


    def __gt__(self, other):
        if not isinstance(other, Rdata) and self.rdclass != other.rdclass or self.rdtype != other.rdtype:
            return NotImplemented

        return self._cmp(other) > 0


    def from_text(cls, rdclass, rdtype, tok, origin = None, relativize = True):
        '''Build an rdata object from text format.

        @param rdclass: The rdata class
        @type rdclass: int
        @param rdtype: The rdata type
        @type rdtype: int
        @param tok: The tokenizer
        @type tok: dns.tokenizer.Tokenizer
        @param origin: The origin to use for relative names
        @type origin: dns.name.Name
        @param relativize: should names be relativized?
        @type relativize: bool
        @rtype: dns.rdata.Rdata instance
        '''
        raise NotImplementedError

    from_text = classmethod(from_text)

    def from_wire(cls, rdclass, rdtype, wire, current, rdlen, origin = None):
        '''Build an rdata object from wire format

        @param rdclass: The rdata class
        @type rdclass: int
        @param rdtype: The rdata type
        @type rdtype: int
        @param wire: The wire-format message
        @type wire: string
        @param current: The offet in wire of the beginning of the rdata.
        @type current: int
        @param rdlen: The length of the wire-format rdata
        @type rdlen: int
        @param origin: The origin to use for relative names
        @type origin: dns.name.Name
        @rtype: dns.rdata.Rdata instance
        '''
        raise NotImplementedError

    from_wire = classmethod(from_wire)

    def choose_relativity(self, origin = None, relativize = True):
        '''Convert any domain names in the rdata to the specified
        relativization.
        '''
        pass



class GenericRdata(Rdata):
    '''Generate Rdata Class

    This class is used for rdata types for which we have no better
    implementation.  It implements the DNS "unknown RRs" scheme.
    '''
    __slots__ = [
        'data']

    def __init__(self, rdclass, rdtype, data):
        super(GenericRdata, self).__init__(rdclass, rdtype)
        self.data = data


    def to_text(self, origin = None, relativize = True, **kw):
        return '\\# %d ' % len(self.data) + _hexify(self.data)


    def from_text(cls, rdclass, rdtype, tok, origin = None, relativize = True):
        if tok.get_string() != '\\#':
            raise dns.exception.SyntaxError, 'generic rdata does not start with \\#'

        length = tok.get_int()
        chunks = []
        while None:
            (ttype, value) = tok.get()
            if ttype == dns.tokenizer.EOL or ttype == dns.tokenizer.EOF:
                break

        hex = ''.join(chunks)
        data = hex.decode('hex_codec')
        if len(data) != length:
            raise dns.exception.SyntaxError, 'generic rdata hex data has wrong length'

        return cls(rdclass, rdtype, data)

    from_text = classmethod(from_text)

    def to_wire(self, file, compress = None, origin = None):
        file.write(self.data)


    def from_wire(cls, rdclass, rdtype, wire, current, rdlen, origin = None):
        return cls(rdclass, rdtype, wire[current:current + rdlen])

    from_wire = classmethod(from_wire)

    def _cmp(self, other):
        return cmp(self.data, other.data)


_rdata_modules = { }
_module_prefix = 'dns.rdtypes'

def get_rdata_class(rdclass, rdtype):

    def import_module(name):
        mod = __import__(name)
        components = name.split('.')
        for comp in components[1:]:
            mod = getattr(mod, comp)

        return mod

    mod = _rdata_modules.get((rdclass, rdtype))
    rdclass_text = dns.rdataclass.to_text(rdclass)
    rdtype_text = dns.rdatatype.to_text(rdtype)
    rdtype_text = rdtype_text.replace('-', '_')
    if not mod:
        mod = _rdata_modules.get((dns.rdatatype.ANY, rdtype))
        if not mod:

            try:
                mod = import_module('.'.join([
                    _module_prefix,
                    rdclass_text,
                    rdtype_text]))
                _rdata_modules[(rdclass, rdtype)] = mod
            except ImportError:

                try:
                    mod = import_module('.'.join([
                        _module_prefix,
                        'ANY',
                        rdtype_text]))
                    _rdata_modules[(dns.rdataclass.ANY, rdtype)] = mod
                except ImportError:
                    mod = None
                except:
                    None<EXCEPTION MATCH>ImportError


                None<EXCEPTION MATCH>ImportError


        None<EXCEPTION MATCH>ImportError

    if mod:
        cls = getattr(mod, rdtype_text)
    else:
        cls = GenericRdata
    return cls


def from_text(rdclass, rdtype, tok, origin = None, relativize = True):
    '''Build an rdata object from text format.

    This function attempts to dynamically load a class which
    implements the specified rdata class and type.  If there is no
    class-and-type-specific implementation, the GenericRdata class
    is used.

    Once a class is chosen, its from_text() class method is called
    with the parameters to this function.

    @param rdclass: The rdata class
    @type rdclass: int
    @param rdtype: The rdata type
    @type rdtype: int
    @param tok: The tokenizer
    @type tok: dns.tokenizer.Tokenizer
    @param origin: The origin to use for relative names
    @type origin: dns.name.Name
    @param relativize: Should names be relativized?
    @type relativize: bool
    @rtype: dns.rdata.Rdata instance'''
    if isinstance(tok, str):
        tok = dns.tokenizer.Tokenizer(tok)

    cls = get_rdata_class(rdclass, rdtype)
    if cls != GenericRdata:
        token = tok.get()
        tok.unget(token)
        if token[0] == dns.tokenizer.IDENTIFIER and token[1] == '\\#':
            rdata = GenericRdata.from_text(rdclass, rdtype, tok, origin, relativize)
            return from_wire(rdclass, rdtype, rdata.data, 0, len(rdata.data), origin)


    return cls.from_text(rdclass, rdtype, tok, origin, relativize)


def from_wire(rdclass, rdtype, wire, current, rdlen, origin = None):
    '''Build an rdata object from wire format

    This function attempts to dynamically load a class which
    implements the specified rdata class and type.  If there is no
    class-and-type-specific implementation, the GenericRdata class
    is used.

    Once a class is chosen, its from_wire() class method is called
    with the parameters to this function.

    @param rdclass: The rdata class
    @type rdclass: int
    @param rdtype: The rdata type
    @type rdtype: int
    @param wire: The wire-format message
    @type wire: string
    @param current: The offet in wire of the beginning of the rdata.
    @type current: int
    @param rdlen: The length of the wire-format rdata
    @type rdlen: int
    @param origin: The origin to use for relative names
    @type origin: dns.name.Name
    @rtype: dns.rdata.Rdata instance'''
    cls = get_rdata_class(rdclass, rdtype)
    return cls.from_wire(rdclass, rdtype, wire, current, rdlen, origin)